package com.atguigu.pro4

import com.atguigu.pro3.OrderEvent
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @description: individual patterns,个体模式可以包括单例模式和循环模式,单例模式只接收一个事件,而循环模式可以接收多个,可以在一个个体模式后追加两次,也就是指定循环次数
 * @time: 2021/4/6 14:30 订单超时事件
 * @author: baojinlong
 * */


// 定义到账事件样例类
case class ReceiptEvent(txId: String, payChannel: String, timestamp: Long)

object TxMatch {
  def main(args: Array[String]): Unit = {
    /**
     * 匹配出现4次 start.times(4)     匹配出现2,3或者4次,并且尽可能多地重复 start.times(2,4).greedy
     * 匹配出现0或者4次 start.times(4).optional 匹配出现1次或多次start.oneOrMore
     * 匹配出现2,3或者4次 start.times(2,4) 匹配出现0次,2次或多次,并且尽可能多地重复 start.timesOrMore(2).optional.greedy
     * 个体模式的条件,每隔模式都需要指定触发条件,作为模式是否接受事件进入的判断依据
     * cep中的个体模式主要通过调用.where or until来指定条件
     * 严格近邻 strict contiguity 宽松relaxed 非确定性宽松紧邻non-deterministic relaxed contiguity
     * 匹配事件的提取
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    // 0.读取订单数据
    val orderEventStream: KeyedStream[OrderEvent, String] = environment.readTextFile("src/main/resources/OrderLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)
      .filter(_.eventType == "pay")
      .keyBy(_.txId)

    // 读取到账事件数据
    val receiptEventStream: KeyedStream[ReceiptEvent, String] = environment.readTextFile("src/main/resources/ReceiptLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        ReceiptEvent(arr(0), arr(1), arr(2).toLong)
      })
      // 升序数据也默认延时1毫秒
      .assignAscendingTimestamps(_.timestamp * 1000)
      .keyBy(_.txId)


    // 合并两条流,进行处理,watermark周期性生成200ms,由于读取文件导致watermark跳变所以输出很多.如果流式输出则不会.
    val resultStream: DataStream[(OrderEvent, ReceiptEvent)] = orderEventStream.connect(receiptEventStream)
      // KeyedCoProcessFunction[K, IN1, IN2, R]
      .process(new TxPayMatchResult)

    resultStream.print("resultStream")
    resultStream.getSideOutput(new OutputTag[OrderEvent]("unmatchedPayEventOutputTag")).print("unmatchedPayEventOutputTag")
    resultStream.getSideOutput(new OutputTag[ReceiptEvent]("unmatchedReceiptEventOutputTag")).print("unmatchedReceiptEventOutputTag")


    environment.execute("order timeout job")
  }

}

// CoProcessFunction也是可以的,以为在上面两个流连接之前已经根据key分组过了
class TxPayMatchResult extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)] {

  // 定义状态,保存当前交易对应的订单支付事件和到账事件
  lazy val payEventState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor("payEventState", classOf[OrderEvent]))
  lazy val receiptEventState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receiptEventState", classOf[ReceiptEvent]))

  // 侧输出流标签
  val unmatchedPayEventOutputTag = new OutputTag[OrderEvent]("unmatchedPayEventOutputTag")
  val unmatchedReceiptEventOutputTag = new OutputTag[ReceiptEvent]("unmatchedReceiptEventOutputTag")

  override def processElement1(pay: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 订单支付事件来了,要判断之前是否有到账事件
    val receipt: ReceiptEvent = receiptEventState.value
    if (receipt != null) {
      // 如果已经有了receipt正常输出匹配,清空状态
      out.collect((pay, receipt))
      receiptEventState.clear()
      payEventState.clear()
    } else {
      // 如果还没来,注册定时器开始等待5秒
      ctx.timerService.registerEventTimeTimer(pay.timestamp * 1000 + 5000)
      // 更新状态
      payEventState.update(pay)
    }
  }

  override def processElement2(receipt: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 到账事件来了,要判断之前是否有pay事件
    val pay: OrderEvent = payEventState.value
    if (pay != null) {
      // 如果已经有了pay,正常输出匹配,清空状态
      out.collect((pay, receipt))
      receiptEventState.clear()
      payEventState.clear()
    } else {
      // 如果还没来,注册定时器开始等待3秒
      ctx.timerService.registerEventTimeTimer(receipt.timestamp * 1000 + 3000)
      receiptEventState.update(receipt)
    }
  }

  // 定时器触发时候逻辑
  override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {
    // 定时器触发,判断状态中哪个还存在,就代表另一个没来,输出到侧输出流
    val orderEvent: OrderEvent = payEventState.value
    if (orderEvent != null) {
      ctx.output(unmatchedPayEventOutputTag, orderEvent)
    }

    val receipt: ReceiptEvent = receiptEventState.value
    if (receipt != null) {
      ctx.output(unmatchedReceiptEventOutputTag, receipt)
    }

    // 清空状态
    receiptEventState.clear()
    payEventState.clear()
  }
}